package cn.lsh.spark.streaming.kafka;

import kafka.serializer.IntegerDecoder;
import kafka.serializer.StringDecoder;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.log4j.lf5.LogLevel;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

public class SparkStreamingOnKafkaDirected {

	public static void main(String[] args) throws InterruptedException {
		SparkConf conf = new SparkConf().setMaster("local").setAppName("SparkStreamingOnKafkaDirected");
		JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
		jsc.sparkContext().setLogLevel(LogLevel.WARN.getLabel());
		Map<String, String> kafkaConf = new HashMap<>();
		kafkaConf.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node00:9092,node01:9092,node02:9092");
		//不能设置earliest，只能设置smallest或largest，smallest就相当于普通kafka里的earliest，
		// 默认largest，只拿实时的数据，即SparkStreaming启动后发到kafka的数据
		kafkaConf.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "smallest");
		Set<String> topics = new HashSet<>();
		topics.add("stream_2");
		JavaPairInputDStream<Integer, String> directStream = KafkaUtils.createDirectStream(jsc,
				Integer.class, //接收key的格式
				String.class, //接受value的格式
				IntegerDecoder.class, //key的反序列化类
				StringDecoder.class, //value的反序列化类
				kafkaConf, topics);
		directStream.transformToPair(rdd -> {
			// ()rdd.rdd();
			return rdd;
		});
		directStream.print(1000);
		jsc.start();
		jsc.awaitTermination();
	}
}
